package com.bw.gmall.realtime.Day0917;



import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.utils.DateFormatUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        //TOdo: 初始化环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // TODO: 2025/9/11 设置并行度为 1
        env.setParallelism(1);

        // TODO: 2025/9/11 状态后端 
        //env.setStateBackend(new HashMapStateBackend());
        //env.enableCheckpointing(5000);

        // TODO: 2025/9/11 从kfk读取主流数据
        String topic = "topic_log";
        String groupId = "base_log_app_group";
        DataStreamSource<String> source = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
        //source.print();

        // TODO: 2025/9/11 数据清洗转换结构  定义一个“侧输出流”的标签（tag），用于标识“脏数据流”
        OutputTag<String> dirtyStreamTag = new OutputTag<String>("dirtyStream") {
        };
        SingleOutputStreamOperator<String> filterStream = source.process(
                new ProcessFunction<String, String>() {
                    @Override
            public void processElement(String value, ProcessFunction<String, String>.Context ctx, Collector<String> out) throws Exception {
                try {
                    // TODO: 2025/9/11 正常情况下，直接通过 out.collect() 输出到主流；
                    out.collect(value);
                }catch (Exception e)
                {
                    // TODO: 2025/9/11 如果发生异常（比如格式不对），则将其发送到侧输出流（side output） 
                    ctx.output(dirtyStreamTag,value);
                    //System.out.println(value);
                }
            }
        });
        //filterStream.print();

        // TODO: 2025/9/11 把字符串转成 JSONObject
        SingleOutputStreamOperator<JSONObject> mappedStream = filterStream.map(JSON::parseObject);

        // TODO: 2025/9/11 按照 mid 字段进行分组  分组后所有属于同一个 mid 的记录会被发往同一个算子实例处理
        KeyedStream<JSONObject, String> keyedStream = mappedStream
                .keyBy(r -> r.getJSONObject("common").getString("mid"));


        SingleOutputStreamOperator<JSONObject> fixedStream = keyedStream.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            // TODO: 2025/9/11 每个 key即每个 mid 都有一个独立的状态变量 firstValueState，保存的是第一次访问的日期字符串
            ValueState<String> firstValueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                firstValueState = getRuntimeContext().getState(new ValueStateDescriptor<String>("firstViewDt", String.class));
            }

            @Override
            public void processElement(JSONObject value, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                String isNew = value.getJSONObject("common").getString("is_new");

                // TODO: 2025/9/11 // 获取当前状态：首次访问日期
                String firstViewDt = firstValueState.value();

                Long ts = value.getLong("ts");

                String dt = DateFormatUtil.toDate(ts);

                // TODO: 2025/9/11 认为是新用户 状态中还没有记录首次访问时间 
                if ("1".equals(isNew)) {
                    if (firstViewDt == null) {
                        firstValueState.update(dt);
                    } else {
                        if (!firstViewDt.equals(dt)) {
                            isNew = "0";
                            value.getJSONObject("common").put("is_new", isNew);
                        }
                    }
                } else {
                    if (firstViewDt == null) {
                        String yesterday = DateFormatUtil.toDate(ts - 1000 * 60 * 60 * 24);
                        firstValueState.update(yesterday);
                    }
                }
                out.collect(value);
            }
        });



        // TODO: 2025/9/11 分流  page  start err
        OutputTag<String> startTag = new OutputTag<String>("startTag"){};
        OutputTag<String> errTag = new OutputTag<String>("errTag"){};
        OutputTag<String> actionTag = new OutputTag<String>("actionTag"){};
        OutputTag<String> displayTag = new OutputTag<String>("displayTag"){};


        SingleOutputStreamOperator<String> separatedStream =
                fixedStream.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, ProcessFunction<JSONObject, String>.Context ctx, Collector<String> out) throws Exception {
                // TODO: 2025/9/11 提取 err 字段：如果存在，则说明是一条错误日志 
                JSONObject error = value.getJSONObject("err");
                if (error != null){
                    // TODO: 2025/9/11 输出到 errTag：将整条原始日志（JSON 字符串）发送到错误流
                    ctx.output(errTag,value.toJSONString());
                }
                value.remove("err");

                // TODO: 2025/9/11 收集启动数据
                JSONObject start = value.getJSONObject("start");
                if (start!=null){
                    ctx.output(startTag,value.toJSONString());
                }
                else {
                    // TODO: 2025/9/11 提取公共字段
                    JSONObject page = value.getJSONObject("page");
                    JSONObject common = value.getJSONObject("common");
                    Long ts = value.getLong("ts");

                    // TODO: 2025/9/11 收集曝光数据[{},{},{}]displays 是一个数组，每个元素是一个曝光事件（如广告、商品展示）；
                    JSONArray displays = value.getJSONArray("displays");
                    if(displays != null){
                        // TODO: 2025/9/11 遍历每个曝光事件，构建新的 JSON 对象
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject display = displays.getJSONObject(i);
                            JSONObject displayObj = new JSONObject();
                            displayObj.put("display", display);
                            displayObj.put("page", page);
                            displayObj.put("common", common);
                            displayObj.put("ts", ts);
                            ctx.output(displayTag, displayObj.toJSONString());
                        }
                        }
                    // TODO: 2025/9/11 收集动作数据
                    JSONArray actions = value.getJSONArray("actions");
                    if(actions != null){
                        for (int i = 0; i < actions.size(); i++) {
                            JSONObject action = actions.getJSONObject(i);
                            JSONObject actionObj = new JSONObject();
                            actionObj.put("action", action);
                            actionObj.put("page", page);
                            actionObj.put("common", common);
                            actionObj.put("ts", ts);
                            ctx.output(actionTag, actionObj.toJSONString());
                }
            }
                    // TODO: 2025/9/11 收集页面数据
                    value.remove("displays");
                    value.remove("actions");

                    out.collect(value.toJSONString());
        }

       }
    });
         separatedStream.print("page>>>");
         separatedStream.getSideOutput(startTag).print("start====>");
         separatedStream.getSideOutput(displayTag).print("display====>");
         separatedStream.getSideOutput(actionTag).print("action====>");
         separatedStream.getSideOutput(errTag).print("error====>");

        // TODO: 2025/9/11 获取侧输出流并写入 Kafka
        DataStream<String> startDS = separatedStream.getSideOutput(startTag);
        DataStream<String> displayDS = separatedStream.getSideOutput(displayTag);
        DataStream<String> actionDS = separatedStream.getSideOutput(actionTag);
        DataStream<String> errorDS = separatedStream.getSideOutput(errTag);

        String page_topic = "dwd_traffic_page_log";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String error_topic = "dwd_traffic_error_log";


        separatedStream.print("page>>>>>>>>>>>>>>>>>>>");
        separatedStream.addSink(MyKafkaUtil.getFlinkKafkaProducer(page_topic));
        startDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(start_topic));
        displayDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(display_topic));
        actionDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(action_topic));
        errorDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(error_topic));
        env.execute();
    }
}